package cn.monkey.state.scheduler.disruptor;

import cn.monkey.state.scheduler.EventPublishScheduler;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.InsufficientCapacityException;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ThreadFactory;

public class RingBufferEventPublisherScheduler implements EventPublishScheduler, EventHandler<EventWrapper<Runnable>> {

    private static final Logger log = LoggerFactory.getLogger(RingBufferEventPublisherScheduler.class);

    private final long id;

    private final Disruptor<EventWrapper<Runnable>> disruptor;

    public RingBufferEventPublisherScheduler(long id,
                                             ThreadFactory threadFactory,
                                             RingBufferConfig ringBufferConfig) {
        this.id = id;
        this.disruptor = new Disruptor<>(EventWrapper::new,
                ringBufferConfig.getBufferSize(),
                threadFactory,
                ringBufferConfig.getProducerType(),
                ringBufferConfig.getWaitStrategySupplier().get());
        this.disruptor.handleEventsWith(this);
    }

    @Override
    public void publish(Runnable event) {
        RingBuffer<EventWrapper<Runnable>> ringBuffer = this.disruptor.getRingBuffer();
        long next;
        try {
            next = ringBuffer.tryNext();
        } catch (InsufficientCapacityException e) {
            if (log.isDebugEnabled()) {
                log.debug("id: {}'s ringBuffer is full, run event directly", this.id());
            }
            event.run();
            return;
        }
        try {
            EventWrapper<Runnable> wrapper = ringBuffer.get(next);
            wrapper.event = event;
        } finally {
            ringBuffer.publish(next);
        }
    }

    @Override
    public long id() {
        return this.id;
    }

    @Override
    public void start() {
        this.disruptor.start();
    }

    @Override
    public boolean isStarted() {
        return this.disruptor.hasStarted();
    }

    @Override
    public void stop() {
        this.disruptor.shutdown();
    }

    @Override
    public void onEvent(EventWrapper<Runnable> event, long sequence, boolean endOfBatch) throws Exception {
        event.event.run();
    }
}
